Use case

use std::time::Duration;

use tokio::sync::mpsc;
use tokio_stream::{StreamExt, wrappers::ReceiverStream};

#[tokio::main]
async fn main() {
    let (tx, rx) = mpsc::channel::<u64>(20);

    let mut receiver = ReceiverStream::new(rx);

    let chunked = (&mut receiver).chunks_timeout(3, Duration::from_secs(3));
    tokio::pin!(chunked);

    tx.send(10).await.unwrap();
    tx.send(20).await.unwrap();

    tokio::select! {
        Some(batch) = chunked.next() => {
            println!("Got: {:?}", batch);
        }
        _ = tokio::time::sleep(Duration::from_secs(1)) => {
            // another condition
        }
    }

    let remainder = chunked.into_remainder();

    // use `receiver` here
}

Thank you @ADD-SP for your review. The original plan was to completely consume the stream indeed and just return the reminder. This however won't work with the Pinned ChunksTimeout stream which is required when polling next() in a select statement.

This means an into_reminder won't actually be useful because the only way for this to be useful is if the ChunksTimeout stream is not pinned. And in that case, the reminder will always guaranteed to be empty since there is no way to interrupt a next().await call (unless you are select statement which requires the stream to be pinned)

I will still rename the method to reminder instead of remaining to match the std interface.

Thank you so much @ADD-SP for your review. I applied all your comments. Let me know if you have more comments.

I think you could add the code example from #7715 (comment) as a snippet in the rustdoc for the new method.

Some new tests would be nice too!

I think you could add the code example from #7715 (comment) as a snippet in the rustdoc for the new method.

Some new tests would be nice too!

@martin-g That example seems too long, and it’s not easy for downstream readers to grasp it in a few seconds. I’m okay with adding an example, but if we do, it should be simpler than the existing one.

Dear @ADD-SP, thank you for your time! I’ve added a test case demonstrating how I would use it mainly to recover any buffered remainder, then restore the inner stream and continue consuming it without losing any items.

The example is a simplified version of the test, though I’m not sure if it’s simple enough for readers. Let me know what you think or if you have any suggestions to improve it

@ADD-SP sorry about changing the title, that was indeed an accident. I use sapling which pushes title and description when submitting the PR. I will use the title you suggested.

Dear @ADD-SP thank you so much for your tips and pointers. I like the tests much more now with the paused-time. Great feature that I have overlooked 😄

Let me know if you still have any more comments.

Thanks!